package com.yanyeori.framework.redis.disruptor.handle;

import com.lmax.disruptor.EventHandler;
import com.yanyeori.framework.redis.config.ConfigFactory;
import com.yanyeori.framework.redis.disruptor.event.RedisPushEvent;
import com.yanyeori.framework.redis.listener.RedisListenerHandler;
import com.yanyeori.framework.redis.model.PushMessage;
import lombok.extern.slf4j.Slf4j;

/**
 * redis事件处理器（消费者）
 *
 * @author chenkuan 2022/5/9
 */
@Slf4j
public class RedisPushEventHandler implements EventHandler<RedisPushEvent> {

    private final int partition;

    public RedisPushEventHandler(int partition) {
        this.partition = partition;
    }

    @Override
    public void onEvent(RedisPushEvent event, long sequence, boolean endOfBatch) {
        if (event == null || event.getTopic() == null) {
            log.warn("sequence: {}, event or topic is null", sequence);
            return;
        }
        if (partition != getPartition(event.getTopic())) return;
        RedisListenerHandler.getInstance().handle(new PushMessage(event.getTopic(), event.getData()));
    }

    private static int getPartition(String key) {
        return (ConfigFactory.getConfig().getEvent().getHandlerSize() - 1) & hash(key);
    }

    private static int hash(Object key) {
        int h;
        return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
    }
}
